package com.tungse.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class FileSysPrintSQL {


    public static void main(String[] args) {
        //创建执行和环境
        EnvironmentSettings setting = EnvironmentSettings
                .newInstance()
                //指定执行模式， 支持inBatchMode 和 inStreamingMode
                .inStreamingMode()
                .build();
        TableEnvironment tEnv = TableEnvironment.create(setting);

        //创建输入表
        String inTableSql =
            "                  CREATE TABLE file_source(\n" +
            "                    name STRING,\n" +
            "                    age  INT,\n" +
            "                    sex  STRING\n" +
            "                  ) WITH (\n" +
            "                    'connector' = 'filesystem',           -- 必选：指定连接器类型\n" +
            "                    'path' = 'file:///stu.json',           -- 必选：指定路径\n" +
            "                    'format' = 'json',                     -- 必选：文件系统连接器指定 format\n" +
            "                    'json.fail-on-missing-field' = 'false',\n" +
            "                    'json.ignore-parse-errors' = 'true'\n" +
            "                  )";
        tEnv.executeSql(inTableSql);

        //创建输出表
        String outTableSql =
                "                  CREATE TABLE print_sink (\n" +
                "                     sex STRING,\n" +
                "                     cnt BIGINT\n" +
                "                  )WITH (\n" +
                "                   'connector' = 'print'\n" +
                "                  )";
        tEnv.executeSql(outTableSql);

        //业务逻辑
        String execSql =
                "                  INSERT INTO print_sink\n" +
                "                  SELECT sex, count(*) AS cnt\n" +
                "                  FROM file_source GROUP BY sex";

        tEnv.executeSql(execSql);

    }
}
